/**
 * Wihidum - Efficient Broker Hierarchy for Publish Subscribe Middleware
 *
 * License      :   Apache License 2.0
 * Home Page    :   http://code.google.com/p/wihidum/
 * Mailing List :   wihidum@googlegroups.com
 */

package lk.mrt.cse.wihidum.delivery;

import lk.mrt.cse.wihidum.communicator.ClusterRouter;
import lk.mrt.cse.wihidum.communicator.Router;
import lk.mrt.cse.wihidum.communicator.TreeRouter;
import lk.mrt.cse.wihidum.processor.PublicationMessage;
import lk.mrt.cse.wihidum.storage.*;
import lk.mrt.cse.wihidum.util.WSEventingConstants;
import lk.mrt.cse.wihidum.util.WihidumConstants;
import lk.mrt.cse.wihidum.util.ClientConfiguration;
import org.apache.axiom.om.OMAbstractFactory;
import org.apache.axiom.om.OMElement;
import org.apache.axiom.om.OMFactory;
import org.apache.axiom.om.OMNamespace;
import org.apache.axiom.soap.*;
import org.apache.axis2.AxisFault;
import org.apache.axis2.transport.http.HTTPConstants;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.client.OperationClient;
import org.apache.axis2.client.Options;
import org.apache.axis2.client.ServiceClient;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.wsdl.WSDLConstants;
import org.apache.log4j.Logger;
import org.apache.commons.httpclient.HttpClient;

import javax.xml.namespace.QName;
import java.util.ArrayList;
import java.util.HashMap;

/**
 * Thread to handle a single publication
 */
public class PublicationThread
        implements Runnable
{

    private static final Logger logger = Logger.getLogger(PublicationThread.class);
    private PublicationMessage publicationMessage;
    private Router router;
    private HashMap<String, String> brokerMap;
    private HttpClient httpClient;
    private String trace = "#publication#";
    private ArrayList<String> traceElements;
    String routedTrace;

    /**
     * Default constructor
     *
     * @param publicationMessage PublicationMessage
     * @param brokerMap          BrokerMap
     */
    public PublicationThread(PublicationMessage publicationMessage, HashMap<String, String> brokerMap)
    {
        this.publicationMessage = publicationMessage;
        this.brokerMap = brokerMap;
        httpClient = new HttpClient();
        traceElements = new ArrayList<String>();
    }

    /**
     * Send the publication to interested subscribers
     */
    public void run()
    {
        trace = publicationMessage.getMessageID().concat(trace).concat(publicationMessage.getTopic()).concat("#");
        if (brokerMap.get("topology").equals(WihidumConstants.TopologyNames.clusterTopology))
        {
            router = new ClusterRouter();
            routedTrace = router.propagatePublication(publicationMessage);
            traceElements.add(routedTrace);
            logger.info(routedTrace);
            this.deliverPublication();
            this.deliverBrokerPublication();
        }
        if (brokerMap.get("topology").equals(WihidumConstants.TopologyNames.treeTopology))
        {
            router = new TreeRouter();
            this.deliverPublication();
            this.deliverBrokerPublication();
        }
        this.sendTraceMessage();
    }

    private void deliverPublication()
    {
        try
        {
            ServiceClient serviceClient;
            serviceClient = new ServiceClient(ClientConfiguration.getConfigurationContext(), null);

            //Delivering publication to normal subscribers
            ArrayList<Subscriber> subscriberList = SubscriptionMap.getSubscribersFilteredByTopic(publicationMessage.getTopic());

            if (subscriberList != null && !subscriberList.isEmpty())
            {
                for (Subscriber subscriber : subscriberList)
                {
                    //creating in out operating client
                    OperationClient operationClient = serviceClient.createClient(ServiceClient.ANON_OUT_ONLY_OP);

                    //creating message context
                    MessageContext outMessageContext = new MessageContext();
                    Options options = outMessageContext.getOptions();
                    options.setProperty(HTTPConstants.CACHED_HTTP_CLIENT, httpClient);
                    options.setProperty(HTTPConstants.REUSE_HTTP_CLIENT, Boolean.TRUE);

                    String SOAPVersion = options.getSoapVersionURI();
                    if (SOAPVersion == null)
                        SOAPVersion = SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI;

                    EndpointReference targetEPR = new EndpointReference(subscriber.getNotifyToAddress());
                    options.setTo(targetEPR);
                    options.setAction(WihidumConstants.Operations.STORE_PUBLICATION);

                    SOAPEnvelope deliveryEnvelope = this.createPublicationDeliveryMessage(SOAPVersion, subscriber.getSubscriberIdentifier().getClientID());

                    outMessageContext.setEnvelope(deliveryEnvelope);
                    operationClient.addMessageContext(outMessageContext);
                    operationClient.execute(true);
                    traceElements.add("0");
                    logger.debug("Publication delivery process completed successfully");
                    logger.info("Event#" + publicationMessage.getMessageID() + " was routed successfully");                    
                }
            }
        }
        catch (NullPointerException ex)
        {
            logger.error("Error in processing store publication response", ex);
        }
        catch (AxisFault ex)
        {
            logger.error("Error in sending publication", ex);
        }
    }

    private void deliverBrokerPublication()
    {
        try
        {
            ServiceClient serviceClient;
            serviceClient = new ServiceClient(ClientConfiguration.getConfigurationContext(), null);

            //Delivering publication to broker subscribers
            ArrayList<BrokerSubscription> brokerSubscriptionList = BrokerSubscriptionMap.getBrokerSubscribersFilteredByTopic(publicationMessage.getTopic());
            if (brokerSubscriptionList != null && !brokerSubscriptionList.isEmpty())
            {
                for (BrokerSubscription brokerSubscription : brokerSubscriptionList)
                {
                    logger.debug("Preparing to send publication to broker subscriber : " + brokerSubscription.getFromBrokerAddress());
                    if (!brokerSubscription.getFromBrokerAddress().equals(publicationMessage.getBrokerPublicationElementFrom()))
                    {
                        //creating in out operating client
                        OperationClient operationClient = serviceClient.createClient(ServiceClient.ANON_OUT_IN_OP);

                        //creating message context
                        MessageContext outMessageContext = new MessageContext();
                        Options options = outMessageContext.getOptions();

                        options.setProperty(HTTPConstants.CACHED_HTTP_CLIENT, httpClient);
                        options.setProperty(HTTPConstants.REUSE_HTTP_CLIENT, Boolean.TRUE);

                        String SOAPVersion = options.getSoapVersionURI();
                        if (SOAPVersion == null)
                            SOAPVersion = SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI;

                        options.setTo(new EndpointReference(brokerSubscription.getFromBrokerAddress()));
                        options.setAction(WihidumConstants.Operations.BROKER_PUBLISH);

                        NeighbourBroker neighbourBroker = new NeighbourBroker();
                        neighbourBroker.setAddress(brokerSubscription.getFromBrokerAddress());

                        neighbourBroker.setBroekrID(brokerSubscription.getFromBrokerID());

                        SOAPEnvelope brokerPublicationMessage = router.createPublicationRequest(publicationMessage, neighbourBroker, SOAPVersion);

                        outMessageContext.setEnvelope(brokerPublicationMessage);
                        operationClient.addMessageContext(outMessageContext);
                        logger.debug("Sending publication to broker subscriber : " + brokerSubscription.getFromBrokerAddress());
                        operationClient.execute(true);

                        MessageContext inMessageContext = operationClient.getMessageContext(WSDLConstants.MESSAGE_LABEL_IN_VALUE);

                        SOAPEnvelope response = inMessageContext.getEnvelope();
                        traceElements.add(neighbourBroker.getBrokerID());
                        logger.debug("Publication delivery response: " + this.processBrokerPublicationResponse(response));
                        logger.info("Event#" + publicationMessage.getMessageID() + " was routed successfully");
                    } else
                    {
                        logger.debug("This publication has arrived from the same broker subscriber : " + brokerSubscription.getFromBrokerAddress() + " to which we are trying to diliver the broker publication. Hence, the delivery was cancelled.");
                    }
                }
            }
        }
        catch (NullPointerException ex)
        {
            logger.error("Error in processing store publication response", ex);
        }
        catch (AxisFault ex)
        {
            logger.error("Error in sending publication", ex);
        }
    }

    /*
   /**
    * Create element to be sent to the listener
    *
    * @param SOAPURI SOAP version to be used in delivery message
    * @return Delivery element
    */
    private SOAPEnvelope createPublicationDeliveryMessage(String SOAPURI, String subscriberID)
    {
        try
        {
            SOAPFactory soapFactory = null;
            if (SOAP11Constants.SOAP_ENVELOPE_NAMESPACE_URI.equals(SOAPURI))
                soapFactory = OMAbstractFactory.getSOAP11Factory();
            else if (SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI.equals(SOAPURI))
                soapFactory = OMAbstractFactory.getSOAP12Factory();
            else throw new Exception("Unknown SOAP version");

            SOAPEnvelope soapEnvelope = soapFactory.getDefaultEnvelope();

            //declare the used namespaces in the SOAP Envelope
            OMNamespace wseventingNS = soapFactory.createOMNamespace(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.EVENTING_PREFIX);
            soapEnvelope.declareNamespace(wseventingNS);
            OMNamespace wsaNS = soapFactory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ADDRESSING_PREFIX);
            soapEnvelope.declareNamespace(wsaNS);
            OMNamespace wihidumNS = soapFactory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE, WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
            soapEnvelope.declareNamespace(wihidumNS);

            SOAPHeader soapHeader = soapEnvelope.getHeader();

            OMFactory factory = OMAbstractFactory.getOMFactory();


            OMElement publicationDeliveryHeader = factory.createOMElement("PublicationDeliveryHeader", wihidumNS);

            OMElement subscriptionIDElement = factory.createOMElement("SubscriptionID", wihidumNS);
            OMElement publisherAddressElement = factory.createOMElement("PublisherAddress", wihidumNS);
            OMElement topicElement = factory.createOMElement("Topic", wihidumNS);

            publisherAddressElement.setText(publicationMessage.getEventSourceURL());
            topicElement.setText(publicationMessage.getTopic());
            subscriptionIDElement.setText(subscriberID);

            publicationDeliveryHeader.addChild(subscriptionIDElement);
            publicationDeliveryHeader.addChild(publisherAddressElement);
            publicationDeliveryHeader.addChild(topicElement);

            soapHeader.addChild(publicationDeliveryHeader);
            SOAPBody soapBody = soapEnvelope.getBody();
            soapBody.setText(publicationMessage.getBody());

            return soapEnvelope;
        }
        catch (Exception ex)
        {
            logger.error(ex);
            return null;
        }

    }

    //TODO: Attach error channel
    private String processBrokerPublicationResponse(SOAPEnvelope responseEnvelope)
    {
        String responseString = "";
        try
        {
            if (responseEnvelope != null)
            {
                OMElement responseElement = responseEnvelope.getBody().getFirstElement();
                OMElement referenceParameters = responseElement.getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "ReferenceParameters"));
                OMElement sourceEPRElement = referenceParameters.getFirstChildWithName(new QName(WSEventingConstants.WIHIDUM_NAMESPACE, "SourceEPR"));
                responseString = sourceEPRElement.getText();
            } else
                responseString = "Error";
        }
        catch (NullPointerException ex)
        {
            ex.printStackTrace();
            responseString = "Error";
        }
        return responseString;
    }

    private void sendTraceMessage()
    {
        //Building the trace string
        String myId = brokerMap.get("id");
        boolean subscriberTraced = false;
        String messageTrace = "";

        if (publicationMessage.getAction().equals(WihidumConstants.Operations.PUBLISH))
        {
            messageTrace = messageTrace.concat("0-" + myId);
        }

        for (int i = 0; i < traceElements.size(); i++)
        {
            if (traceElements.get(i) != null)
            {
                if (traceElements.get(i).equals("0"))
                {
                    if (!subscriberTraced)
                    {
                         if (!messageTrace.isEmpty())
                            messageTrace = messageTrace.concat("," + myId + "-0");
                         else
                            messageTrace = messageTrace.concat(myId + "-0");
                    }
                    subscriberTraced = true;
                } else
                {
                    if (!messageTrace.isEmpty())
                        messageTrace = messageTrace.concat("," + myId + "-" + traceElements.get(i));
                    else
                        messageTrace = messageTrace.concat(myId + "-" + traceElements.get(i));
                }
                logger.info("Publication trace:" +messageTrace);
            }
        }

        if (!messageTrace.isEmpty())
        {
            trace = trace.concat(messageTrace);
            logger.info("Sending publication trace from broker: "+trace);
            try
            {
                ServiceClient serviceClient = new ServiceClient(ClientConfiguration.getConfigurationContext(), null);

                //creating in out operating client
                OperationClient operationClient = serviceClient.createClient(ServiceClient.ANON_OUT_IN_OP);

                //creating message context
                MessageContext outMessageContext = new MessageContext();
                Options options = outMessageContext.getOptions();
                options.setProperty(HTTPConstants.CACHED_HTTP_CLIENT, httpClient);
                options.setProperty(HTTPConstants.REUSE_HTTP_CLIENT, Boolean.TRUE);

                String SOAPVersion = options.getSoapVersionURI();
                if (SOAPVersion == null)
                    SOAPVersion = SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI;

                EndpointReference targetEPR = new EndpointReference(brokerMap.get("admin"));
                options.setTo(targetEPR);
                options.setAction(WihidumConstants.Operations.SET_MESSAGE_TRACE);

                SOAPEnvelope traceEnvelope = this.createTraceMessage(SOAPVersion);

                outMessageContext.setEnvelope(traceEnvelope);
                operationClient.addMessageContext(outMessageContext);
                operationClient.execute(true);
                logger.info("SetTrace message was sent successfully");
            } catch (AxisFault ex)
            {
                logger.error("Error in sending SetTrace message", ex);
            }
            catch (NullPointerException ex)
            {
                logger.error("Error in SetTrace message", ex);
            }
        }
    }

    private SOAPEnvelope createTraceMessage(String SOAPURI)
    {
        try
        {
            SOAPFactory soapFactory = null;
            if (SOAP11Constants.SOAP_ENVELOPE_NAMESPACE_URI.equals(SOAPURI))
                soapFactory = OMAbstractFactory.getSOAP11Factory();
            else if (SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI.equals(SOAPURI))
                soapFactory = OMAbstractFactory.getSOAP12Factory();
            else throw new Exception("Unknown SOAP version");

            SOAPEnvelope soapEnvelope = soapFactory.getDefaultEnvelope();

            //declare the used namespaces in the SOAP Envelope
            OMNamespace wseventingNS = soapFactory.createOMNamespace(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.EVENTING_PREFIX);
            soapEnvelope.declareNamespace(wseventingNS);
            OMNamespace wsaNS = soapFactory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ADDRESSING_PREFIX);
            soapEnvelope.declareNamespace(wsaNS);
            OMNamespace wihidumNS = soapFactory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE, WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
            soapEnvelope.declareNamespace(wihidumNS);

            OMFactory factory = OMAbstractFactory.getOMFactory();

            OMElement distributionTraceElement = factory.createOMElement("DistributionTrace", wihidumNS);           
            distributionTraceElement.setText(trace);

            SOAPBody soapBody = soapEnvelope.getBody();
            soapBody.addChild(distributionTraceElement);

            return soapEnvelope;
        }
        catch (Exception ex)
        {
            logger.error("Error in creating the SetTrace message"+ex);
            return null;
        }
    }

}

